package com.gblfy.kafka.controller;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class KafkaAsyncController {
    private final static Logger log = LoggerFactory.getLogger(KafkaAsyncController.class);

    @Autowired
    private KafkaTemplate<Integer, String> kafkaTemplate;

    //设置回调函数，异步等待broker端的返回结束
    @RequestMapping("/send/async/{message}")
    public String sendAsync(@PathVariable String message) {

        //同步发送消息
        ListenableFuture<SendResult<Integer, String>> future = kafkaTemplate.send("topic-springboot-01", 0, 1, message);
        future.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {
            @Override
            public void onFailure(Throwable e) {
                log.info("发送消息失败: {}", e.getMessage());
            }

            @Override
            public void onSuccess(SendResult<Integer, String> result) {
                RecordMetadata metadata = result.getRecordMetadata();
                log.info("发送的主题:{} ,发送的分区:{} ,发送的偏移量:{}  ",
                        metadata.topic(), metadata.partition(), metadata.offset());
            }
        });
        return "success";
    }
}
